package com.redis.delay;

import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.Map;
import java.util.concurrent.*;
import java.util.function.BiConsumer;

/**
 * @Description
 * @Author lizhilong
 * @Date 2020/12/4 19:36
 */

@Slf4j
public class RedisDelayedQueueInit implements ApplicationContextAware {
    @Resource
    private RedissonClient redssionClient;

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {

        Map<String, RedissDelayQueueCallback> beansOfType = applicationContext.getBeansOfType(RedissDelayQueueCallback.class);
        ScheduledExecutorService executorService = Executors.newScheduledThreadPool(4);
        beansOfType.forEach(new BiConsumer<String, RedissDelayQueueCallback>() {
            @Override
            public void accept(String s, RedissDelayQueueCallback redissDelayQueueCallback) {
                String name = redissDelayQueueCallback.getClass().getName();
                addListenerThread(name, redissDelayQueueCallback, executorService);
            }
        });

    }


    private <T> void addListenerThread(String queueName, RedissDelayQueueCallback redissDelayQueueCallback, ScheduledExecutorService executorService) {
        log.info("添加监听线程，队列名称：{}", queueName);
        RBlockingQueue<T> blockingFairQueue = redssionClient.getBlockingQueue(queueName);
        executorService.scheduleWithFixedDelay(new Runnable() {
            @Override
            public void run() {
                log.info("定时线程扫描一次：{}", LocalDateTime.now());
                try {
                    T take = blockingFairQueue.take();
                    log.info("队列：{}，获取到：{}", queueName, JSONUtil.parse(take).toString());
                    new Thread(() -> {
                        redissDelayQueueCallback.callback(take);
                    }).start();
                } catch (InterruptedException e) {
                    log.error(e.getMessage());
                }
            }
        }, 1, 2, TimeUnit.SECONDS);
    }
}
